
#include "../aprtools/APRAutoMemoryPool.h"
#include "Broker5Client.h"

/** Do a BrokerMessage exchange by sending it and receiving another.
 * 
 * @param[in,out] msg Message to send to the broker. Is replaced by
 * the received message
 * 
 * @param kdb The key database to use to authenticate and encrypt it.
 * 
 * @return The BrokerMessage that is received. */
BrokerMessage *Broker5Client::exchangeMessage(BrokerMessage &msg, KeyDB &kdb)
{
   int retval = sendMessage(msg, kdb);
   msg.reset();
   if (retval == 1) { errprint("error sending message"); return NULL; }
   retval = receiveMessage(msg, kdb);
   if (retval == 1) { errprint("error receiving message"); return NULL; }
   return &msg;
}

/** A threading wrapper around the message exchange.
 * 
 * @param tthread The APR thread.
 * 
 * @param args A void pointer to some wrapped arguments.
 * 
 * @return A void pointer to a wrapped return value. */
void *Broker5Client::exchangeMessageThread(apr_thread_t *tthread, void *args)
{
   Broker5Client *kthis = (Broker5Client *)((void **)args)[0];
   BrokerMessage *msg = (BrokerMessage *)((void **)args)[1];
   KeyDB *kdb = (KeyDB *)((void **)args)[2];
   BrokerMessage **nret = new (BrokerMessage *)();
   *nret = kthis->exchangeMessage(*msg, *kdb);
   return nret;
}

/** Do a BrokerMessage exchange by sending it and receiving
 * another. The BrokerMessage returned is the same BrokerMessage that
 * was passed in, with the data overwritten with the response. The
 * APRFuture must also be deleted. This is a non-blocking call because
 * of the future.
 * 
 * @param msg Message to send to the broker.
 * 
 * @param kdb The key database to use to encrypt and sign the outgoing
 * message and to decrypt and authenticate the response.
 * 
 * @return The BrokerMessage that is received. */
APRFuture<BrokerMessage *> *Broker5Client::exchangeMessageFuture(BrokerMessage &msg, KeyDB &kdb)
{
   APRFuture<BrokerMessage *> *exchange_message_future = new APRFuture<BrokerMessage *>(mp);
   exchange_message_future->addArgument((void *)this);
   exchange_message_future->addArgument((void *)(&msg));
   exchange_message_future->addArgument((void *)(&kdb));
   exchange_message_future->addFunctor(exchangeMessageThread);
   exchange_message_future->start();
   return exchange_message_future;
}

int Broker5Client::sendMessage(BrokerMessage &msg, KeyDB &kdb)
{
   // return 1 for failure
   if (!isvalid) { errprint("network problem"); return 1; }
   APRBucketBrigade *aprbb = new APRBucketBrigade(mp);
   int send_err = msg.sendEncryptedOnBucketBrigade(aprbb, &kdb);
   int success = (send_err == 0);
   if (success)
   {
      aprbb->writeEOS();
      apr_size_t rlen = aprbb->realLength();
      apr_size_t sentbytes = sendBucketBrigade(aprbb);
      int retval = (sentbytes != rlen);
      if (retval)
         errprint("expected to send %" APR_SIZE_T_FMT " bytes, actually sent %" APR_SIZE_T_FMT, sentbytes, rlen);
      delete aprbb;
      return retval;
   } else
   {
      errprint("sending message did not succeed");
      delete aprbb;
      return 1;
   }
}

/** Receive a BrokerMessage into the class. The BrokerMessage must be
 * deleted after use.
 * 
 * @param msg Message to send to the broker.
 * 
 * @param kdb The key database to use to authenticate and encrypt it.
 * 
 * @return The BrokerMessage that is received. */
int Broker5Client::receiveMessage(BrokerMessage &msg, KeyDB &kdb)
{
   int recv_err = msg.receiveEncryptedOnBucketBrigade(clientbb, &kdb, NULL);
   int success = (recv_err == 0);
   if (success)
   {
      // Divine meaning from the BrokerMessage message!
   }
   return 0;
}

/** A function for managing the network exchange on the client
 * side. This usually involves exchanging a message asking something
 * of the broker, for a message containing a status or error code.
 * 
 * @param bmsg The broker message that is to be sent to the broker,
 * and subsequently replaced by a reply.
 * 
 * @param imp The memory pool to use for the exchange.
 * 
 * @param cancelflag A pointer to a piece of memory that is dirty-read
 * to provide an instant cancel to an external managing thread.
 * 
 * @param thostname The host name to contact.
 * 
 * @param tportno The port number to contact.
 * 
 * @param tkeydb The key database to use for the cryptography on the
 * connection.
 * 
 * @param ttimeout The timeout for the connection.
 * 
 * @return One if the exchange was cancelled or otherwise cut short,
 * zero on success. */
int Broker5Client::exchangeMessage(BrokerMessage &bmsg, apr_pool_t *imp, int *cancelflag, const char *thostname, int tportno, KeyDB &tkeydb, int ttimeout)
{
   int cancelled = 0;
   APRAutoMemoryPool pool(imp);
   Broker5Client *brokerc = new Broker5Client(thostname, tportno, pool.mp());
   if (brokerc->isValid() == 0)
   {
      errprint("client has not connected");
      delete brokerc;
      return 1;
   }
   APRFuture<BrokerMessage *> *future_response = brokerc->exchangeMessageFuture(bmsg, tkeydb);
   int timeout = 1;
   while (!(future_response->isFinished()))
   {
      apr_sleep(timeout);
      timeout = (timeout > ttimeout) ? ttimeout : (timeout <<= 1);
      if (cancelflag)
         if (*cancelflag)
         {
            errprint("cancelled while waiting for message exchange to complete");
            cancelled = 1;
            break;
         }
   }
   if (cancelled)
   {
      future_response->cancel();
      delete future_response;
      bmsg.reset();
   } else
   {
      future_response->get();
      delete future_response;
   }
   future_response = NULL;
   delete brokerc;
   return cancelled;
}
